package dao

import org.apache.spark.sql.{DataFrame, SaveMode}
import util.JDBCUtil
import java.sql.DriverManager
import org.slf4j.LoggerFactory

object DAO {
  private val logger = LoggerFactory.getLogger(getClass)

  // 从配置文件或环境变量获取数据库连接信息
  def getJdbcConfig(): (String, String, String) = {
    val jdbcUrl = sys.env.getOrElse("JDBC_URL", "jdbc:mysql://localhost:3306/your_database")
    val username = sys.env.getOrElse("DB_USER", "your_username")
    val password = sys.env.getOrElse("DB_PASSWORD", "your_password")
    (jdbcUrl, username, password)
  }

  // 测试数据库连接
  def testConnection(): Boolean = {
    var connection: java.sql.Connection = null
    try {
      connection = JDBCUtil.getConnection
      logger.info("数据库连接测试成功")
      true
    } catch {
      case e: Exception =>
        logger.error(s"数据库连接失败: ${e.getMessage}", e)
        false
    } finally {
      if (connection != null) connection.close()
    }
  }

  // 保存分析结果到数据库
  def saveAnalysisResult(df: DataFrame, tableName: String, saveMode: SaveMode = SaveMode.Overwrite): Unit = {
    val (jdbcUrl, username, password) = getJdbcConfig()
    val connectionProperties = new java.util.Properties()
    connectionProperties.put("user", username)
    connectionProperties.put("password", password)

    // 设置数据库驱动
    val dbType = jdbcUrl.split(":")(1).toLowerCase()
    dbType match {
      case "mysql" => connectionProperties.put("driver", "com.mysql.cj.jdbc.Driver")
      case "postgresql" => connectionProperties.put("driver", "org.postgresql.Driver")
      case "sqlserver" => connectionProperties.put("driver", "com.microsoft.sqlserver.jdbc.SQLServerDriver")
      case _ => logger.warn("未知数据库类型，使用默认驱动")
    }

    // 验证数据
    if (df.isEmpty) {
      logger.warn("数据集为空，不会写入数据库")
      return
    }

    try {
      // 配置写入选项
      df.write
        .mode(saveMode)
        .option("batchsize", "1000")  // 每批插入1000条记录
        .option("isolationLevel", "READ_COMMITTED")  // 事务隔离级别
        .jdbc(jdbcUrl, tableName, connectionProperties)

      logger.info(s"数据已成功保存到 $tableName 表，共 ${df.count()} 条记录")
    } catch {
      case e: Exception =>
        logger.error(s"保存数据到 $tableName 表时出错: ${e.getMessage}", e)
        throw new RuntimeException("数据库写入失败", e)
    }
  }
}